package server

import (
	proto "github.com/golang/protobuf/proto"
	"go.uber.org/zap"
	"gse/protos"
	"sync"
)

const DefaultEventPushFrequency = 20

type EventConfig struct {
	PushFrequency uint16
}

type EventPusherConfig struct {
	///
	/// 默认的固定推送频率: 每秒推送的事件数。如: freq等于20时,则每隔50毫秒推送一次, 一次可推送多个消息/事件.
	///
	DefaultEventPushFrequency uint16
}

type EventPusher struct {
	cfg *EventPusherConfig
	// map[eventname]*map[*Session]*map[seriesID] *EventPushReg
	eventPushRegs *sync.Map
	// map[eventname]EventConfig
	registrations *sync.Map
}

func NewEventPusher(cfg *EventPusherConfig) *EventPusher {
	if cfg.DefaultEventPushFrequency == 0 {
		cfg.DefaultEventPushFrequency = DefaultEventPushFrequency
	}
	return &EventPusher{
		cfg:           cfg,
		eventPushRegs: new(sync.Map),
		registrations: new(sync.Map),
	}
}

func (ep *EventPusher) OnNewSub(ses *Session, iSub proto.Message) proto.Message {
	sub := iSub.(*protos.SubEvent)
	rsp := new(protos.SubEventResponse)
	cfgI, ok := ep.registrations.Load(sub.Name)
	logger().Debug("registering new remote event subscription.", zap.String("name", sub.Name))
	if !ok {
		rsp.Status = protos.EventRpcStatusTypes_Failed
		rsp.Msg = "event " + sub.Name + " not register."
		logger().Warn("try register illegal event.", zap.String("name", sub.Name))
		return rsp
	}
	cfg := cfgI.(*EventConfig)
	var seriesID = ses.genSubSeriesID()
	rsp.SeriesID = seriesID
	regs := new(sync.Map)
	oldRegs, ok := ep.eventPushRegs.LoadOrStore(sub.Name, regs)
	if ok {
		regs = oldRegs.(*sync.Map)
	}
	regSes := new(sync.Map)
	oldRegSes, ok := regs.LoadOrStore(ses, regSes)
	if ok {
		regSes = oldRegSes.(*sync.Map)
	}

	reg := NewEventPushReg(&EventPushRegConfig{
		EventPushFrequency: cfg.PushFrequency,
		Ses:                ses,
		SeriesID:           uint8(seriesID),
	})

	regSes.Store(seriesID, reg)
	regs.Store(ses, regSes)
	rsp.SeriesID = seriesID
	rsp.Status = protos.EventRpcStatusTypes_OK
	logger().Debug("registered new remote event subscription.", zap.String("name", sub.Name), zap.Uint32("seriesID", rsp.SeriesID))
	ses.cfg.RegisterHook(OnClose(func(ses *Session, args ...interface{}) {
		regSes.Delete(seriesID)
		reg.Stop()
	}))
	return rsp
}

func (ep *EventPusher) OnUnSub(iUnSub proto.Message) proto.Message {
	unSub := iUnSub.(*protos.CancelSubEvent)
	rsp := new(protos.CancelSubEventResponse)
	subI, ok := ep.eventPushRegs.Load(unSub.SeriesID)
	logger().Debug("canceling remote event subscription.", zap.Uint32("seriesID", unSub.SeriesID))
	if !ok {
		rsp.Status = protos.EventRpcStatusTypes_Failed
		rsp.Msg = "subscription not exists."
		logger().Warn("try cancel unregistered event.", zap.Uint32("seriesID", unSub.SeriesID))
		return rsp
	}
	sub := subI.(*EventPushReg)
	sub.Stop()
	ep.eventPushRegs.Delete(unSub.SeriesID)
	rsp.Status = protos.EventRpcStatusTypes_OK
	logger().Debug("remote event subscription canceled.", zap.Uint32("seriesID", unSub.SeriesID))
	return rsp
}

func (ep *EventPusher) RegisterEvent(name string, cfg *EventConfig) {
	if cfg.PushFrequency < 1 {
		cfg.PushFrequency = ep.cfg.DefaultEventPushFrequency
	}
	ep.registrations.Store(name, cfg)
}

func (ep *EventPusher) Stop() {
	ep.eventPushRegs.Range(func(k, v interface{}) bool {
		v.(*sync.Map).Range(func(k, v interface{}) bool {
			v.(*sync.Map).Range(func(k, v interface{}) bool {
				v.(*EventPushReg).Stop()
				return true
			})
			return true
		})
		return true
	})
}

func (ep *EventPusher) QueueEventToSession(name string, ses *Session, evt proto.Message) {
	segsI, ok := ep.eventPushRegs.Load(name)
	if !ok {
		return
	}
	segs := segsI.(*sync.Map)
	sesI, ok := segs.Load(ses)
	if !ok {
		return
	}
	sesReg := sesI.(*sync.Map)
	sesReg.Range(func(k, v interface{}) bool {
		v.(*EventPushReg).QueueEvent(evt)
		return true
	})
}

func (ep *EventPusher) CreateGroup(name string) *EventPushGroup {
	g := &EventPushGroup{
		Name:     name,
		sessions: new(sync.Map),
		puser:    ep,
	}
	return g
}
